package com.atguigu.actual.edu0417.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.actual.edu0417.app.BaseApp;
import com.atguigu.actual.edu0417.beans.DwsTradePaymentSucBean;
import com.atguigu.actual.edu0417.common.EduConfig;
import com.atguigu.actual.edu0417.unils.DateFormatUtil;
import com.atguigu.actual.edu0417.unils.DorisSinkUtil;
import com.atguigu.actual.edu0417.unils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author: techyang
 * @since: 2023-10-22 10:52
 * @description: 支付成功人数统计
 **/
public class DwsTradePaymentSucWindow extends BaseApp {

    public static void main(String[] args) {
        DwsTradePaymentSucWindow dwsTradePaymentSucWindow = new DwsTradePaymentSucWindow();
        dwsTradePaymentSucWindow.base(args);
    }

    @Override
    public void process(StreamExecutionEnvironment env) {
        // 读取支付成功的数据
        DataStreamSource<String> kafkaStrDs = env.fromSource(MyKafkaUtil.getKafkaSource(EduConfig.KAFKA_DWD_TRADE_PAY_DETAIL_SUC, "dws_trade_payment_suc_group")
                , WatermarkStrategy.noWatermarks(), "kafka_source");

        // 转换为jsonObj
        SingleOutputStreamOperator<DwsTradePaymentSucBean> resultDs = kafkaStrDs.map(JSONObject::parseObject)
                // 分配水位线
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                                .withTimestampAssigner(
                                        (jsonObj, ts) -> jsonObj.getLong("ts") * 1000
                                )
                )
                // 按照用户分组
                .keyBy(jsonObj -> jsonObj.getString("user_id")).process(new ProcessFunction<JSONObject, JSONObject>() {

                    private ValueState<String> lastPaySucState;


                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("lastPaySucState", String.class);
                        // 设置状态失效时间为1天
                        stateDescriptor.enableTimeToLive(new StateTtlConfig.Builder(Time.days(1)).build());
                        lastPaySucState = getRuntimeContext().getState(stateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        // 上次支付成功日期
                        String lastPaySucDate = lastPaySucState.value();
                        String curPaySucDate = DateFormatUtil.toDate(jsonObj.getLong("ts") * 1000);
                        if (StringUtils.isEmpty(lastPaySucDate) || !curPaySucDate.equals(lastPaySucDate)) {
                            // 当日第一次支付成功
                            lastPaySucState.update(curPaySucDate);
                            out.collect(jsonObj);
                        }
                    }
                })

                // 开窗
                .windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(5)))
                // 聚合统计
                .aggregate(new AggregateFunction<JSONObject, Long, Long>() {
                               @Override
                               public Long createAccumulator() {
                                   return 0L;
                               }

                               @Override
                               public Long add(JSONObject value, Long accumulator) {
                                   return ++accumulator;
                               }

                               @Override
                               public Long getResult(Long accumulator) {
                                   return accumulator;
                               }

                               @Override
                               public Long merge(Long a, Long b) {
                                   return null;
                               }
                           },
                        new AllWindowFunction<Long, DwsTradePaymentSucBean, TimeWindow>() {
                            @Override
                            public void apply(TimeWindow window, Iterable<Long> values, Collector<DwsTradePaymentSucBean> out) throws Exception {
                                long start = window.getStart();
                                long end = window.getEnd();
                                String stt = DateFormatUtil.toYmdHms(start);
                                String edt = DateFormatUtil.toYmdHms(end);
                                String curDate = DateFormatUtil.toDate(start);
                                for (Long count : values) {
                                    out.collect(new DwsTradePaymentSucBean(stt, edt, curDate, count));
                                }
                            }
                        });

        resultDs.print("resultDs>>>");
        // 写入doris
        resultDs.addSink(DorisSinkUtil.getJdbcSink("insert into dws_trade_payment_suc_window values (?,?,?,?)"));

    }
}
